package com.wis.storm_test;

import com.wis.bolt.common.*;
import com.wis.bolt.mpg.*;
import com.wis.bolt.risk.*;
import com.wis.bolt.to_kafka.*;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.apache.storm.Config;
import org.apache.storm.LocalCluster;
import org.apache.storm.StormSubmitter;
import org.apache.storm.generated.AlreadyAliveException;
import org.apache.storm.generated.AuthorizationException;
import org.apache.storm.generated.InvalidTopologyException;
import org.apache.storm.kafka.spout.KafkaSpout;
import org.apache.storm.kafka.spout.KafkaSpoutConfig;
import org.apache.storm.topology.TopologyBuilder;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.io.InputStream;
import java.util.Date;
import java.util.Properties;

/**
 * @Description: traffic_fan    主函数
 * @author: fan
 * @Date: Created in 2018/10/9 14:33
 * @Modified By:
 */
public class TopologyDrive {
    public static void main(String[] args) {
        TopologyBuilder builder = new TopologyBuilder();
        builder.setSpout("TestSpout", new LocalSpout(), 1);
        builder.setBolt("PACSplitBolt", new TestBolt(), 3).allGrouping("TestSpout");

        Config config = new Config();
        config.put(Config.TOPOLOGY_DEBUG, false);
        config.setNumWorkers(1);
        if (args.length < 1) {
            new LocalCluster().submitTopology("1", config, builder.createTopology());
        }
    }
}